package com.lndsnx.demo;

import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConfirmCallback;
import com.lndsnx.config.AmqpConfig;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.UUID;
import java.util.concurrent.ConcurrentNavigableMap;
import java.util.concurrent.ConcurrentSkipListMap;

/**
 * @Description 生产者
 * @Author lndsnx
 * @Date 2025/7/30 上午9:50
 **/
public class Producer {
    // 队列名称
    private static final String QUEUE_NAME = "queue-test";
    // 交换机名称
    private static final String EXCHANGE_NAME = "exchange-test";
    // 路由名称
    private static final String ROUTING_KEY = "key_test";
    // 虚拟主机名称
    private static final String VHOST = "vhost_test";
    // 公网接入点
    private static final String HOSTNAME = AmqpConfig.HOSTNAME;
    // 账号
    private static final String USERNAME = AmqpConfig.USERNAME;
    // 密码
    private static final String PASSWORD = AmqpConfig.PASSWORD;
    //如果需要使用 TLS 加密，改为 5671 端口，并且enableSSL设置为true。
    public static final int PORT = 5672;
    public static final boolean enableSSL = false;

    // 通道
    private Channel channel;
    // 待确认的消息
    private final ConcurrentNavigableMap<Long/*deliveryTag*/, String/*msgId*/> outstandingConfirms;
    // 连接工厂
    private final ConnectionFactory factory;

    public Producer(ConnectionFactory factory) throws Exception {
        this.factory = factory;
        this.outstandingConfirms = new ConcurrentSkipListMap<>();
        this.channel = factory.createChannel();
    }

    public static void main(String[] args) throws Exception {
        // 构建连接工厂。
        ConnectionFactory factory = new ConnectionFactory(HOSTNAME, PORT, USERNAME, PASSWORD, VHOST, enableSSL);

        // 初始化生产者。
        Producer producer = new Producer(factory);

        // declare。
        producer.declare();

        // 初始化通道配置。
        producer.initChannel();

        for (int i = 0; i < 10; i++) {
            // 发送消息。
            producer.doSend("hello,amqp");
            Thread.sleep(100);
        }
    }

    /**
     * 初始化通道配置
     */
    private void initChannel() throws IOException {
        // 确认模式开启
        channel.confirmSelect();

        // 添加ack确认监听器。
        ConfirmCallback cleanOutstandingConfirms = (deliveryTag, multiple) -> {
            // 如果是批量确认
            if (multiple) {
                // 获取所有小于等于当前 deliveryTag 的待确认消息
                ConcurrentNavigableMap<Long, String> confirmed = outstandingConfirms.headMap(deliveryTag, true);

                // 遍历并打印确认消息信息
                for (Long tag : confirmed.keySet()) {
                    String msgId = confirmed.get(tag);
                    System.out.format("消息msgId : %s 已确认, tag: %d, multiple: %b%n", msgId, tag, true);
                }

                // 清空已确认的消息
                confirmed.clear();
            } else {
                // 单条确认，移除并获取确认消息的 msgId
                String msgId = outstandingConfirms.remove(deliveryTag);
                System.out.format("消息msgId : %s 已确认, tag: %d, multiple: %b%n", msgId, deliveryTag, false);
            }
        };
        // 添加确认监听器，处理确认和未确认消息
        channel.addConfirmListener(cleanOutstandingConfirms, (deliveryTag, multiple) -> {
            // 获取未确认消息的 msgId
            String msgId = outstandingConfirms.get(deliveryTag);
            System.err.format("消息msgId : %s 未确认, tag: %d, multiple: %b%n", msgId, deliveryTag, multiple);
        });

        // 添加返回监听器，处理无法路由的消息
        channel.addReturnListener(returnMessage -> System.out.println("return msgId=" + returnMessage.getProperties().getMessageId()));
    }

    /**
     * 声明交换机、队列并绑定
     */
    private void declare() throws IOException {
        // 创建一个 direct 类型的持久化交换机
        channel.exchangeDeclare(EXCHANGE_NAME, "direct", true);
        // 创建一个持久化队列
        channel.queueDeclare(QUEUE_NAME, true, false, false, null);
        // 绑定队列和交换机
        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, ROUTING_KEY);
    }

    /**
     * 发送消息
     */
    private void doSend(String content) throws Exception {
        try {
            // 生成唯一的消息ID
            String msgId = UUID.randomUUID().toString();
            // 构建消息属性，设置消息ID
            AMQP.BasicProperties props = new AMQP.BasicProperties.Builder().messageId(msgId).build();

            // 发布消息到交换机
            channel.basicPublish(EXCHANGE_NAME, ROUTING_KEY, true, props, content.getBytes(StandardCharsets.UTF_8));

            // 将消息添加到待确认映射表中
            outstandingConfirms.put(channel.getNextPublishSeqNo(), msgId);
        } catch (Exception e) {
            //need reconnect if channel is closed.
            String message = e.getMessage();

            System.out.println(message);

            // 判断是否需要重连
            if (channelClosedByServer(message)) {
                // 关闭当前通道
                factory.closeCon(channel);
                // 创建新通道
                channel = factory.createChannel();
                // 重新初始化通道
                this.initChannel();
                // 重新发送消息
                doSend(content);
            } else {
                throw e;
            }
        }
    }

    /**
     * 判断通道是否因服务器内部错误而关闭
     */
    private boolean channelClosedByServer(String errorMsg) {
        // 检查错误信息是否包含特定的服务器内部错误标识
        if (errorMsg != null
                && errorMsg.contains("channel.close")
                && errorMsg.contains("reply-code=541")
                && errorMsg.contains("reply-text=InternalError")) {
            return true;
        } else {
            return false;
        }
    }
}

